package com.demo.java.disruptor.netty.client;


import com.demo.java.disruptor.netty.common.TranslatorData;
import com.demo.java.disruptor.netty.common.AbstractConsumer;
import com.demo.java.disruptor.netty.common.DisruptorFactory;
import com.demo.java.disruptor.netty.common.MarshallingCodeCFactory;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    private ChannelFuture channelFuture;
    private Channel channel;
    private NioEventLoopGroup workerGroup;
    private Bootstrap bootstrap;
    public NettyClient(ProducerType producerType, int bufferSize, WaitStrategy waitStrategy, int consumerSize) {
        this.workerGroup = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(workerGroup)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.TCP_NODELAY, true)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                                .addLast(MarshallingCodeCFactory.buildMarshallingDecoder())
                                .addLast(MarshallingCodeCFactory.buildMarshallingEncoder())
                                .addLast(new ClientMessageHandler());
                    }
                });
        AbstractConsumer[] consumers = new ClientConsumer[consumerSize];
        for (int i = 0; i < consumerSize; i++) {
            consumers[i] = DisruptorFactory.createClientConsumer();
        }
        DisruptorFactory.getInstance().init(producerType, bufferSize, waitStrategy, consumers);
    }
    
    public void start() {
        try {
            this.channelFuture = this.bootstrap.connect("localhost", 9000).sync();
            this.channel = this.channelFuture.channel();
            for (int i = 0; i < 1000000; i++) {
                System.out.println("client send msg");
                TranslatorData translatorData = new TranslatorData();
                translatorData.setName("客户端名称" + i);
                translatorData.setMsg("客户端消息" + i);
                this.channel.writeAndFlush(translatorData);
            }
            this.channel.closeFuture().sync();
            System.out.println("end");
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void stop() {
        try {
            workerGroup.shutdownGracefully();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
